#include "../Common/Net.hpp"
#include "../Common/Dispatcher.hpp"
#include "../Common/Message.hpp"
#include "../Client/Requestor.hpp"
#include <unordered_set>

namespace zgwrpc
{
    namespace client
    {
        class TopicManager
        {
        public:
            using ptr = std::shared_ptr<TopicManager>;
            using Callback = std::function<void(const std::string &key, const std::string &msg)>; // 针对订阅的主题的回调函数
            TopicManager(Requestor::ptr requestor): _requestor(requestor) {};
            // 创建主题
            bool Create(const BaseConnection::ptr &conn, const std::string &topicname)
            {
                // 1.构建请i去
                return CommonRequest(conn, topicname, TopicOpType::TOPIC_CREATE);
            }
            // 删除主题
            bool Remove(const BaseConnection::ptr &conn, const std::string &topicname)
            {
                return CommonRequest(conn, topicname, TopicOpType::TOPIC_REMOVE);
            }
            // 订阅主题
            bool Subscribe(const BaseConnection::ptr &conn, const std::string &topicname, const Callback &callback)
            {
                _callbacks.insert({topicname, callback});
                // 先建立订阅关系
                bool ret = CommonRequest(conn, topicname, TopicOpType::TOPIC_SUBSCRIBE);
                if (!ret)
                    RemoveCallback(topicname);
                return ret;
            }
            void AddCallback(const std::string &topicname, const Callback &callback)
            {
                std::lock_guard<std::mutex> lock(_mutex);
                _callbacks.insert({topicname, callback});
            }
            void RemoveCallback(const std::string &topicname)
            {
                std::lock_guard<std::mutex> lock(_mutex);
                _callbacks.erase(topicname);
            }

            // 取消订阅主题
            bool Cancel(const BaseConnection::ptr &conn, const std::string &topicname)
            {
                RemoveCallback(topicname);
                return CommonRequest(conn, topicname, TopicOpType::TOPIC_CANCEL);
            }
            //发送消息
            bool OnPublish(const BaseConnection::ptr &conn, const std::string &topicname, const std::string &msg)
            {
                return CommonRequest(conn, topicname, TopicOpType::TOPIC_PUBLISH, msg);
            }

            bool CommonRequest(const BaseConnection::ptr &conn, const std::string &topicname, TopicOpType type, const std::string &msg = "")
            {
                auto msg_req = MessageFactory::create<TopicRequest>();
                msg_req->setTopicOpType(type);
                msg_req->setId(Uuid::Getid());
                msg_req->setTopicKey(topicname);
                msg_req->setMType(MType::REQ_TOPIC);
                if (type == TopicOpType::TOPIC_PUBLISH)
                    msg_req->setTopicMsg(msg);
                // 2.发送请求
                BaseMessage::ptr msg_rsp;

                int ret = _requestor->Send(conn, msg_req, msg_rsp);
                if (ret == false)
                    LOG(LogLevel::ERROR) << "创建主题失败";

                TopicResponse::ptr top_rsp_msg = dynamic_pointer_cast<TopicResponse>(msg_rsp);
                if (top_rsp_msg.get() == nullptr)
                {
                    LOG(LogLevel::ERROR) << "TOPIC响应消息类型错误";
                    return false;
                }
                if (top_rsp_msg->rCode() != RCode::RCODE_OK)
                {
                    LOG(LogLevel::ERROR) << "TOPIC响应消息错误码:";
                    return false;
                }
                return true;
            }
            void OnPublishMessage(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg)
            {
                // 这里边只处理消息推送
                TopicOpType type = msg->topicOpType();
                if (type != TopicOpType::TOPIC_PUBLISH)
                {
                    LOG(LogLevel::ERROR) << "非法的TOPIC消息类型";
                    return;
                }

                string topicname = msg->topicKey();
                string msg_str = msg->topicMsg();
                auto callback = GetCallback(topicname);
                if (callback)
                {
                    callback(topicname, msg_str);
                }
                else 
                {
                    LOG(LogLevel::ERROR) << "没有找到对应的回调函数";
                }
            }
            const Callback GetCallback(const std::string &topicname)
            {
                std::lock_guard<std::mutex> lock(_mutex);
                auto it = _callbacks.find(topicname);
                if (it == _callbacks.end())
                    return Callback();
                return it->second;
            }

        private:
            std::mutex _mutex;
            std::unordered_map<std::string, Callback> _callbacks;
            Requestor::ptr _requestor;
        };
    }
}